package com.example;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * Streaming WordCount Example
 * 
 * 这是一个基于流处理的Flink程序示例，演示如何：
 * 1. 创建流执行环境
 * 2. 从本地文件中读取流数据源
 * 3. 使用滚动窗口进行聚合
 * 4. 实时输出统计结果
 */
public class StreamingWordCount {

    public static void main(String[] args) throws Exception {
        // 获取流执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 设置并行度为1，便于观察输出
        env.setParallelism(1);

        String inputPath = "data/input.txt";
        
        if (args.length > 0) {
            inputPath = args[0];
        }

        // 从文本文件创建数据流源
        DataStream<String> text = readTextFileAsStream(env, inputPath);

        // 处理数据：分割单词、统计
        @SuppressWarnings("deprecation")
        DataStream<Tuple2<String, Integer>> wordCounts = text
                // FlatMap: 将每行文本分割成单词
                .flatMap(new Tokenizer())
                // 按第一个字段（单词）分组
                .keyBy(value -> value.f0)
                // 使用5秒的滚动窗口
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                // 对窗口内的数据求和
                .sum(1);

        // 打印结果
        wordCounts.print();

        // 执行任务
        env.execute("Streaming WordCount Example");
    }

    /**
     * 从文本文件读取数据作为流
     * 使用 @SuppressWarnings 处理已弃用的 API
     */
    @SuppressWarnings("deprecation")
    private static DataStream<String> readTextFileAsStream(StreamExecutionEnvironment env, String filePath) {
        return env.readTextFile(filePath);
    }

    /**
     * 实现 FlatMapFunction，用于分割单词
     */
    public static class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
        @Override
        public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
            String[] tokens = value.toLowerCase().split("\\W+");
            
            for (String token : tokens) {
                if (token.length() > 0) {
                    out.collect(new Tuple2<>(token, 1));
                }
            }
        }
    }
}
